package io.kiki.sba.registry.server.clientInterface.bootstrap;


import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.metrics.ReporterUtils;
import io.kiki.sba.registry.metrics.TaskMetrics;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.server.clientInterface.metadata.MetadataCacheRegistry;
import io.kiki.sba.registry.server.clientInterface.providedata.ConfigProvideDataWatcher;
import io.kiki.sba.registry.server.clientInterface.remoting.handler.ClientNodeConnectionHandler;
import io.kiki.sba.registry.server.clientInterface.slot.SlotsCache;
import io.kiki.sba.registry.server.clientInterface.strategy.RegistryStrategy;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.server.shared.providedata.SystemPropertyProcessorManager;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import io.kiki.sba.registry.store.api.meta.ClientManagerAddressRepository;
import io.kiki.sba.registry.store.api.meta.RecoverConfigRepository;
import org.glassfish.jersey.server.ResourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;

import javax.annotation.Resource;
import javax.ws.rs.Path;
import javax.ws.rs.ext.Provider;
import java.lang.annotation.Annotation;
import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


public class Bootstrap {

    private static final Logger logger = LoggerFactory.getLogger(Bootstrap.class);
    private final AtomicBoolean metaStart = new AtomicBoolean(false);
    private final AtomicBoolean schedulerStart = new AtomicBoolean(false);
    private final AtomicBoolean httpStart = new AtomicBoolean(false);
    private final AtomicBoolean consoleStart = new AtomicBoolean(false);
    private final AtomicBoolean serverStart = new AtomicBoolean(false);
    private final AtomicBoolean dataStart = new AtomicBoolean(false);
    private final AtomicBoolean serverForSessionSyncStart = new AtomicBoolean(false);
    private final Retryer<Boolean> startupRetryer = RetryerBuilder.<Boolean>newBuilder().retryIfRuntimeException().retryIfResult(input -> !input).withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(10)).build();
    private final Retryer<Boolean> addBlacklistRetryer = RetryerBuilder.<Boolean>newBuilder().retryIfException().retryIfRuntimeException().withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(3)).build();
    @Autowired
    protected MetaServerService metaNodeService;
    @Autowired
    private Config config;
    @Autowired
    private DataExchanger boltDataExchanger;
    @Autowired
    private DataExchanger jerseyDataExchanger;
    @Autowired
    private ExecutorManager executorManager;
    @Resource(name = "serverHandlers")
    private Collection<AbstractServerHandler> serverHandlers;
    @Autowired
    private NodeExchanger dataNodeExchanger;
    @Autowired
    private NodeExchanger dataNodeNotifyExchanger;
    @Autowired
    private ResourceConfig jerseyResourceConfig;
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private SystemPropertyProcessorManager systemPropertyProcessorManager;
    @Autowired
    private SlotsCache slotsCache;
    @Autowired
    private ConfigProvideDataWatcher configProvideDataWatcher;
    @Autowired
    private RegistryStrategy registryStrategy;
    @Autowired
    private ClientManagerAddressRepository clientManagerAddressRepository;
    @Autowired
    private RecoverConfigRepository recoverConfigRepository;
    private Server server;
    private Server dataSyncServer;
    @Resource(name = "sessionSyncHandlers")
    private Collection<AbstractServerHandler> sessionSyncHandlers;
    private Server consoleServer;
    @Resource(name = "consoleHandlers")
    private Collection<AbstractServerHandler> consoleHandlers;
    @Autowired
    private MetadataCacheRegistry metadataCacheRegistry;
    @Resource
    private ClientNodeConnectionHandler clientNodeConnectionHandler;
    private Server httpServer;


    public void start() {
        try {
            logger.info("release properties: {}", ServerEnv.getReleaseProps());
            logger.info("the configuration items are as follows: " + config.toString());

            initEnvironment();
            ReporterUtils.enablePrometheusDefaultExports();

            openSessionSyncServer();

            startupRetryer.call(() -> {
                connectMetaServer();
                return true;
            });

            // wait until slot table is get
            startupRetryer.call(() -> slotsCache.getLocalSlots().getEpoch() != Slots.INIT.getEpoch());

            metadataCacheRegistry.startSynced();

            recoverConfigRepository.waitSynced();
            metadataCacheRegistry.waitSynced();
            clientManagerAddressRepository.waitSynced();

            startupRetryer.call(() -> systemPropertyProcessorManager.startFetchPersistenceSystemProperty());

            startScheduler();

            openHttpServer();

            startupRetryer.call(() -> {
                connectDataServer();
                return true;
            });

            registryStrategy.start();
            configProvideDataWatcher.start();

            openConsoleServer();
            openSessionServer();

            TaskMetrics.getInstance().registerBolt();

            logger.info("Initialized Session Server...");
            postStart();
            Runtime.getRuntime().addShutdownHook(new Thread(this::doStop));
        } catch (Throwable e) {
            logger.error("Cannot bootstrap session server :", e);
            throw new RuntimeException("Cannot bootstrap session server :", e);
        }
    }


    public void destroy() {
        doStop();
        Runtime.getRuntime().halt(0);
    }

    private void postStart() throws Throwable {
        startupRetryer.call(() -> {
            metaNodeService.removeSelfFromMetaBlacklist();
            logger.info("successful start session server, remove self from blacklist");
            return true;
        });
    }

    private void doStop() {
        try {
            logger.info("{} Shutting down Session Server..", new Date().toString());
            stopHttpServer();
            clientNodeConnectionHandler.stop(); // stop process disconnect event
            stopServer();
            // stop http server and client bolt server before add blacklist
            // make sure client reconnect to other sessions and data
            gracefulShutdown();
            stopDataSyncServer();
            stopConsoleServer();
            executorManager.stopScheduler();
        } catch (Throwable e) {
            logger.error("Shutting down Session Server error!", e);
        }
        logger.info("{} Session server is now shutdown...", new Date().toString());
    }

    private void initEnvironment() {
        logger.info("Session server Environment: DataCenter {},Region {},ProcessId {}", config.getDataCenter(), config.getSessionServerRegion(), ServerEnv.PROCESS_ID);
    }

    private void startScheduler() {

        try {
            if (schedulerStart.compareAndSet(false, true)) {
                executorManager.startScheduler();
                logger.info("Session Scheduler started!");
            }
        } catch (Exception e) {
            schedulerStart.set(false);
            logger.error("Session Scheduler start error!", e);
            throw new RuntimeException("Session Scheduler start error!", e);
        }
    }

    private void openConsoleServer() {
        try {
            if (consoleStart.compareAndSet(false, true)) {
                consoleServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), config.getConsolePort()), 1024 * 128, 1024 * 256, consoleHandlers.toArray(new ChannelHandler[consoleHandlers.size()]));
                logger.info("Console server started! port:{}", config.getConsolePort());
            }
        } catch (Exception e) {
            serverStart.set(false);
            logger.error("Console server start error! port:{}", config.getConsolePort(), e);
            throw new RuntimeException("Console server start error!", e);
        }
    }

    private void openSessionServer() {
        try {
            if (serverStart.compareAndSet(false, true)) {
                server = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), config.getServerPort()), config.getClientIOLowWaterMark(), config.getClientIOHighWaterMark(), serverHandlers.toArray(new ChannelHandler[serverHandlers.size()]));
                logger.info("Session server started! port:{}", config.getServerPort());
            }
        } catch (Exception e) {
            serverStart.set(false);
            logger.error("Session server start error! port:{}", config.getServerPort(), e);
            throw new RuntimeException("Session server start error!", e);
        }
    }

    private void openSessionSyncServer() {
        try {
            if (serverForSessionSyncStart.compareAndSet(false, true)) {
                dataSyncServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), config.getSyncSessionPort()), config.getSyncSessionIOLowWaterMark(), config.getSyncSessionIOHighWaterMark(), sessionSyncHandlers.toArray(new ChannelHandler[sessionSyncHandlers.size()]));
                logger.info("Data server for sync started! port:{}", config.getSyncSessionPort());
            }
        } catch (Exception e) {
            serverForSessionSyncStart.set(false);
            logger.error("Data sync server start error! port:{}", config.getSyncSessionPort(), e);
            throw new RuntimeException("Data sync server start error!", e);
        }
    }

    private void connectDataServer() {
        try {
            dataNodeNotifyExchanger.connectServer();
            dataNodeExchanger.connectServer();
            dataStart.set(true);
        } catch (Exception e) {
            dataStart.set(false);
            logger.error("Data server connected server error! port:{}", config.getDataServerPort(), e);
            throw new RuntimeException("Data server connected server error!", e);
        }
    }

    private void connectMetaServer() {
        try {
            // register node as renew node
            metaNodeService.renewNode();
            // start sched renew
            metaNodeService.startRenewer();

            // start fetch system property data
            startupRetryer.call(() -> systemPropertyProcessorManager.startFetchMetaSystemProperty());

            metaStart.set(true);

            logger.info("MetaServer connected meta server! Port:{}", config.getMetaServerPort());
        } catch (Exception e) {
            metaStart.set(false);
            logger.error("MetaServer connected server error! Port:{}", config.getMetaServerPort(), e);
            throw new RuntimeException("MetaServer connected server error!", e);
        }
    }

    private void openHttpServer() {
        try {
            if (httpStart.compareAndSet(false, true)) {
                bindResourceConfig();
                httpServer = jerseyDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), config.getHttpServerPort()), new ResourceConfig[]{jerseyResourceConfig});
                logger.info("Open http server port {} success!", config.getHttpServerPort());
            }
        } catch (Exception e) {
            logger.error("Open http server port {} error!", config.getHttpServerPort(), e);
            httpStart.set(false);
            throw new RuntimeException("Open http server error!", e);
        }
    }

    private void bindResourceConfig() {
        registerInstances(Path.class);
        registerInstances(Provider.class);
    }

    private void registerInstances(Class<? extends Annotation> annotationType) {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(annotationType);
        if (beans != null && !beans.isEmpty()) {
            beans.forEach((beanName, bean) -> {
                jerseyResourceConfig.registerInstances(bean);
                jerseyResourceConfig.register(bean.getClass());
            });
        }
    }


    private void stopServer() {
        if (server != null && server.isOpen()) {
            server.close();
        }
    }

    private void stopDataSyncServer() {
        if (dataSyncServer != null && dataSyncServer.isOpen()) {
            dataSyncServer.close();
        }
    }

    private void stopConsoleServer() {
        if (consoleServer != null && consoleServer.isOpen()) {
            consoleServer.close();
        }
    }

    private void stopHttpServer() {
        if (httpServer != null && httpServer.isOpen()) {
            httpServer.close();
        }
    }

    private void gracefulShutdown() {
        if (!config.isGracefulShutdown()) {
            logger.info("disable graceful shutdown, skip add blacklist");
            return;
        }
        try {
            addBlacklistRetryer.call(() -> {
                logger.info("[GracefulShutdown] add self to blacklist");
                metaNodeService.addSelfToMetaBlacklist();
                return true;
            });
            logger.info("add session self to blacklist successfully");
        } catch (Throwable e) {
            logger.error("add session self to blacklist failed", e);
        }
    }


    public boolean getMetaStart() {
        return metaStart.get();
    }


    public boolean getSchedulerStart() {
        return schedulerStart.get();
    }


    public boolean getHttpStart() {
        return httpStart.get();
    }


    public boolean getServerStart() {
        return serverStart.get();
    }

    public boolean getDataStart() {
        return dataStart.get();
    }

    public boolean getConsoleStart() {
        return consoleStart.get();
    }

    public boolean getServerForSessionSyncStart() {
        return serverForSessionSyncStart.get();
    }
}
